#include "raft/raft.h"

#include "zrpc/base/thread.h"
#include "zrpc/base/thread_pool.h"

namespace libraft {
const std::string Raft::kMemberConfigKey = "#MEMBERCONFIG";
const std::string Raft::kSnapShotPath = "snap_shot/";
const std::string Raft::kCheckPointDir = "data_check_point/";

static const int32_t kTimeOut = 60;
static const int32_t kThreadCount = 8;
static const int32_t kLRUCache = 1024 * 64 * 64;
static const int32_t kBloomFilter = 10;
static const int32_t kWriteBufferSize = 1024 * 64 * 64;
static const int32_t kSnapshotCount = 100;

Raft::Raft(const Options& options)
    : options_(options),
      apply_db_(nullptr),
      log_db_(nullptr),
      meta_db_(nullptr),
      loop_(new zrpc::EventLoop()) {}

Raft::~Raft() {}

static void BuildMembership(const std::vector<std::string>& opt_members,
                            raft::Membership* members) {
  members->Clear();
  for (const auto& m : opt_members) {
    members->add_nodes(m);
  }
}

static void BuildAddServerRequest(const std::string& add_server,
                                  raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kAddServer);
  raft::CmdRequest_AddServerRequest* add_server_request =
      cmd->mutable_add_server_request();
  add_server_request->set_new_server(add_server);
}

static void BuildRemoveServerRequest(const std::string& remove_server,
                                     raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kRemoveServer);
  raft::CmdRequest_RemoveServerRequest* remove_server_request =
      cmd->mutable_remove_server_request();
  remove_server_request->set_old_server(remove_server);
}

static void BuildReadRequest(const std::string& key, raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kRead);
  raft::CmdRequest_KvRequest* kv_request = cmd->mutable_kv_request();
  kv_request->set_key(key);
}

static void BuildAppendRequest(const std::string& key, const std::string& value,
                               raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kAppend);
  raft::CmdRequest_KvRequest* kv_request = cmd->mutable_kv_request();
  kv_request->set_key(key);
  kv_request->set_value(value);
}

static void BuildWriteRequest(const std::string& key, const std::string& value,
                              raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kWrite);
  raft::CmdRequest_KvRequest* kv_request = cmd->mutable_kv_request();
  kv_request->set_key(key);
  kv_request->set_value(value);
}

static void BuildReadResponse(const std::string& key, const std::string& value,
                              raft::StatusCode code, raft::CmdResponse* res) {
  res->set_code(code);
  raft::CmdResponse_KvResponse* kv_res = res->mutable_kv_response();
  if (code == raft::StatusCode::kOk) {
    kv_res->set_value(value);
  }
}

static void BuildGetAllServersRequest(raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kGetAllServers);
}

static void BuildDeleteRequeset(const std::string& key, raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kDelete);
  raft::CmdRequest_KvRequest* kv_requeset = cmd->mutable_kv_request();
  kv_requeset->set_key(key);
}

static void BuildTryLockRequest(const std::string& name,
                                const std::string& holder, uint64_t ttl,
                                raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kTryLock);
  raft::CmdRequest_LockRequest* lock_request = cmd->mutable_lock_request();
  lock_request->set_name(name);
  lock_request->set_holder(holder);
  lock_request->set_lease_end(rocksdb::Env::Default()->NowMicros() +
                              ttl * 1000);
}

static void BuildUnLockRequest(const std::string& name,
                               const std::string& holder,
                               raft::CmdRequest* cmd) {
  cmd->set_type(raft::Type::kUnLock);
  raft::CmdRequest_LockRequest* lock_request = cmd->mutable_lock_request();
  lock_request->set_name(name);
  lock_request->set_holder(holder);
}

static void BuildLogEntry(const raft::CmdRequest& cmd,
                          const uint64_t current_term, raft::Entry* entry) {
  entry->set_term(current_term);
  entry->set_key(cmd.kv_request().key());
  entry->set_value(cmd.kv_request().value());

  if (cmd.type() == raft::Type::kRead) {
    entry->set_optype(raft::Entry_OpType_kRead);
  } else if (cmd.type() == raft::Type::kWrite) {
    entry->set_optype(raft::Entry_OpType_kWrite);
  } else if (cmd.type() == raft::Type::kDelete) {
    entry->set_optype(raft::Entry_OpType_kDelete);
  } else if (cmd.type() == raft::Type::kTryLock) {
    entry->set_optype(raft::Entry_OpType_kTryLock);
    entry->set_key(cmd.lock_request().name());
    entry->set_holder(cmd.lock_request().holder());
    entry->set_lease_end(cmd.lock_request().lease_end());
  } else if (cmd.type() == raft::Type::kUnLock) {
    entry->set_optype(raft::Entry_OpType_kUnLock);
    entry->set_key(cmd.lock_request().name());
    entry->set_holder(cmd.lock_request().holder());
  } else if (cmd.type() == raft::Type::kAddServer) {
    entry->set_optype(raft::Entry_OpType_kAddServer);
    entry->set_server(cmd.add_server_request().new_server());
  } else if (cmd.type() == raft::kRemoveServer) {
    entry->set_optype(raft::Entry_OpType_kRemoveServer);
    entry->set_server(cmd.remove_server_request().old_server());
  } else if (cmd.type() == raft::kGetAllServers) {
    entry->set_optype(raft::Entry_OpType_kGetAllServers);
  } else if (cmd.type() == raft::Type::kAppend) {
    entry->set_optype(raft::Entry_OpType_kAppend);
  }
}

static bool IgnoreBuildLog(const raft::Entry& entry) {
  if (entry.optype() == raft::Entry_OpType_kAppend) {
    return false;
  }
  return true;
}

rocksdb::Status Raft::ExecuteCommand(const raft::CmdRequest& req,
                                     raft::CmdResponse* res) {
  raft::Entry entry;
  BuildLogEntry(req, context_->current_term, &entry);
  res->set_type(req.type());
  res->set_code(raft::StatusCode::kError);

  uint64_t last_log_index = raft_log_->Append(entry);
  if (last_log_index <= 0) {
    return rocksdb::Status::IOError("append entry failed");
  }

  bool ignore = false;
  if (options_.single_mode) {
    context_->commit_index = last_log_index;
    meta_->SetCommitIndex(context_->commit_index);
    apply_->ScheduleApply();
  } else {
    ignore = IgnoreBuildLog(entry);
    if (ignore) {
      primary_->AddTask(kNewCommand);
    }
  }

  {
    std::unique_lock<std::mutex> lk(context_->apply_mutex_);
    while (ignore && context_->last_applied < last_log_index) {
      if (context_->apply_cond_.wait_for(lk, std::chrono::seconds(kTimeOut)) ==
          std::cv_status::timeout) {
        return rocksdb::Status::TimedOut("execute command timeout");
      }
    }

    std::string value;
    rocksdb::Status s;
    raft::Lock lock;

    switch (req.type()) {
      case raft::Type::kAppend: {
        res->set_code(raft::StatusCode::kOk);
        break;
      }

      case raft::Type::kAddServer: {
        res->set_code(raft::StatusCode::kOk);
        break;
      }

      case raft::Type::kRemoveServer: {
        res->set_code(raft::StatusCode::kOk);
        break;
      }

      case raft::Type::kWrite: {
        res->set_code(raft::StatusCode::kOk);
        break;
      }

      case raft::Type::kDelete: {
        res->set_code(raft::StatusCode::kOk);
        break;
      }

      case raft::Type::kTryLock: {
        s = apply_db_->Get(rocksdb::ReadOptions(), req.lock_request().name(),
                           &value);
        if (s.ok()) {
          lock.ParseFromString(value);
          if (lock.holder() == req.lock_request().holder() &&
              lock.lease_end() == req.lock_request().lease_end()) {
            res->set_code(raft::StatusCode::kOk);
          }
        } else {
          res->set_code(raft::StatusCode::kLocked);
        }
        break;
      }

      case raft::Type::kUnLock: {
        s = apply_db_->Get(rocksdb::ReadOptions(), req.lock_request().name(),
                           &value);
        if (s.IsNotFound()) {
          res->set_code(raft::StatusCode::kOk);
        } else {
          res->set_code(raft::StatusCode::kLocked);
        }
        break;
      }

      case raft::Type::kRead: {
        s = apply_db_->Get(rocksdb::ReadOptions(), req.kv_request().key(),
                           &value);
        if (s.ok()) {
          BuildReadResponse(req.kv_request().key(), value,
                            raft::StatusCode::kOk, res);
        } else if (s.IsNotFound()) {
          BuildReadResponse(req.kv_request().key(), value,
                            raft::StatusCode::kNotFound, res);
        } else {
          BuildReadResponse(req.kv_request().key(), value,
                            raft::StatusCode::kError, res);
          return rocksdb::Status::Corruption("get key failed");
        }
        break;
      }

      case raft::Type::kGetAllServers: {
        s = apply_db_->Get(rocksdb::ReadOptions(), kMemberConfigKey, &value);
        if (!s.ok()) {
          return s;
        }

        if (!res->mutable_all_servers()->ParseFromString(value)) {
          return rocksdb::Status::Corruption("parse failed");
        }
        res->set_code(raft::StatusCode::kOk);
        break;
      }
    }
  }
  return rocksdb::Status::OK();
}

rocksdb::Status Raft::RedirectToLeader(const raft::CmdRequest& req,
                                       raft::CmdResponse* res,
                                       const std::string& server_addr) {
  for (auto& it : peers_) {
    if (it->ServerAddr() == server_addr) {
      // return it->SendRedirectToLeader(req, res);
    }
  }
  return rocksdb::Status::Corruption("redirect leader failed");
}

rocksdb::Status Raft::GetLogEntry(const int64_t log_index, std::string& value) {
  raft::Entry entry;
  if (raft_log_->GetEntry(log_index, &entry)) {
    value = entry.value();
    return rocksdb::Status::OK();
  } else {
    return rocksdb::Status::NotFound();
  }
}

rocksdb::Status Raft::Append(const std::string& key, const std::string& value,
                             bool is_leader_log) {
  raft::CmdRequest cmd;
  BuildAppendRequest(key, value, &cmd);
  raft::CmdResponse res;
  rocksdb::Status s = DoCommand(cmd, &res, is_leader_log);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    return rocksdb::Status::OK();
  }
  return rocksdb::Status::Corruption("append failed");
}

rocksdb::Status Raft::Write(const std::string& key, const std::string& value) {
  raft::CmdRequest cmd;
  BuildWriteRequest(key, value, &cmd);
  raft::CmdResponse res;
  rocksdb::Status s = DoCommand(cmd, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    return rocksdb::Status::OK();
  }
  return rocksdb::Status::Corruption("write failed");
}

rocksdb::Status Raft::DoCommand(const raft::CmdRequest& req,
                                raft::CmdResponse* res, bool is_leader_node) {
  std::string leader_ip;
  int leader_port;
  {
    std::unique_lock<std::mutex> lk(context_->global_mutex_);
    leader_ip = context_->leader_ip;
    leader_port = context_->leader_port;
  }

  if (!is_leader_node ||
      (options_.local_ip == leader_ip && options_.local_port == leader_port)) {
    return ExecuteCommand(req, res);
  } else if (leader_ip == "" || leader_port == 0) {
    return rocksdb::Status::Incomplete("no leader node!");
  } else {
    return rocksdb::Status::NotSupported("no redirect to leader");
    // redirect to leader
    // return RedirectToLeader(req, res,
    //                        leader_ip + ":" + std::to_string(leader_port));
  }
}

rocksdb::Status Raft::Read(const std::string& key, std::string* value) {
  raft::CmdRequest req;
  BuildReadRequest(key, &req);
  raft::CmdResponse res;
  rocksdb::Status s = DoCommand(req, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    *value = res.kv_response().value();
    return rocksdb::Status::OK();
  } else if (res.code() == raft::StatusCode::kNotFound) {
    return rocksdb::Status::NotFound("not found the key");
  } else {
    return rocksdb::Status::Corruption("read key failed");
  }
}

rocksdb::Status Raft::Delete(const std::string& key) {
  raft::CmdRequest req;
  raft::CmdResponse res;
  BuildDeleteRequeset(key, &req);

  rocksdb::Status s = DoCommand(req, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    return rocksdb::Status::OK();
  }
  return rocksdb::Status::Corruption("delete failed");
}

rocksdb::Status Raft::DirtyRead(const std::string& key, std::string* value) {
  rocksdb::Status s = apply_db_->Get(rocksdb::ReadOptions(), key, value);
  if (s.ok()) {
    return rocksdb::Status::OK();
  } else if (s.IsNotFound()) {
    return rocksdb::Status::NotFound("");
  } else {
    return rocksdb::Status::Corruption(s.ToString());
  }
}

rocksdb::Status Raft::TryLock(const std::string& name,
                              const std::string& holder, uint64_t ttl) {
  raft::CmdRequest req;
  BuildTryLockRequest(name, holder, ttl, &req);
  raft::CmdResponse res;
  rocksdb::Status s = DoCommand(req, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    return rocksdb::Status::OK();
  }
  return rocksdb::Status::Corruption("lock failed");
}

rocksdb::Status Raft::UnLock(const std::string& name,
                             const std::string& holder) {
  raft::CmdRequest req;
  BuildUnLockRequest(name, holder, &req);
  raft::CmdResponse res;
  rocksdb::Status s = DoCommand(req, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    return rocksdb::Status::OK();
  }
  return rocksdb::Status::Corruption("unlock failed");
}

Options Raft::GetOptions() const { return options_; }

void Raft::Quit() {
  peer_threads_.clear();
  rpc_server_.reset();
  peers_.clear();
  context_.reset();
  primary_.reset();
  raft_log_.reset();
  meta_.reset();
  apply_.reset();

  primary_thread_.reset();
  apply_thread_.reset();
  apply_db_.reset();
  log_db_.reset();
  meta_db_.reset();

  loop_->Quit();
}

void Raft::QuitInLoop() { loop_->RunInLoop(std::bind(&Raft::Quit, this)); }

rocksdb::Status Raft::Destory() {
  rocksdb::Status s = DestroyDB(path_ + options_.db_path, db_options_);
  if (!s.ok()) {
    return s;
  }

  s = DestroyDB(path_ + options_.log_path, db_options_);
  if (!s.ok()) {
    return s;
  }

  s = DestroyDB(path_ + options_.meta_path, db_options_);
  if (!s.ok()) {
    return s;
  }
  return rocksdb::Status::OK();
}

rocksdb::Status Raft::Open(const std::string& path) {
  rocksdb::Options ops;
  ops.create_if_missing = true;
  ops.write_buffer_size = kWriteBufferSize;
  // ops.compaction_filter_factory.reset(new RaftFilterFactory());
  rocksdb::BlockBasedTableOptions table_ops;
  table_ops.block_cache = rocksdb::NewLRUCache(kLRUCache);
  table_ops.filter_policy.reset(
      rocksdb::NewBloomFilterPolicy(kBloomFilter, true));
  ops.table_factory.reset(rocksdb::NewBlockBasedTableFactory(table_ops));

  rocksdb::Env::Default()->CreateDir(path);

  path_ = path;
  db_options_ = ops;

  if (options_.destory) {
    rocksdb::Status s = Destory();
    if (!s.ok()) {
      return s;
    }
  }

  rocksdb::DB* apply_db;
  rocksdb::Status s =
      rocksdb::DB::Open(db_options_, path + options_.db_path, &apply_db);
  if (!s.ok()) {
    return rocksdb::Status::Corruption("open apply_db failed, " + s.ToString());
  }

  apply_db_ = std::shared_ptr<rocksdb::DB>(apply_db);

  rocksdb::DB* log_db;
  s = rocksdb::DB::Open(db_options_, path + options_.log_path, &log_db);
  if (!s.ok()) {
    return rocksdb::Status::Corruption("open log_db failed, " + s.ToString());
  }

  log_db_ = std::shared_ptr<rocksdb::DB>(log_db);

  rocksdb::DB* meta_db;
  s = rocksdb::DB::Open(db_options_, path + options_.meta_path, &meta_db);
  if (!s.ok()) {
    return rocksdb::Status::Corruption("open meta_db failed, " + s.ToString());
  }

  meta_db_ = std::shared_ptr<rocksdb::DB>(meta_db);

  raft_log_.reset(new RaftLog(log_db_));
  meta_.reset(new Meta(meta_db_));
  meta_->Open();

  context_.reset(new Context(options_));
  context_->RecoverInit(meta_);
  // 用于测试节点断开，领导选举, 任期增加
  context_->current_term += options_.add_current_term;

  std::string val;
  raft::Membership do_members;
  s = apply_db_->Get(rocksdb::ReadOptions(), kMemberConfigKey, &val);
  if (!options_.single_mode && s.ok() && do_members.ParseFromString(val)) {
    for (int i = 0; i < do_members.nodes_size(); i++) {
      context_->members_.insert(do_members.nodes(i));
    }
  } else {
    BuildMembership(options_.members, &do_members);
    if (!do_members.SerializeToString(&val)) {
      return rocksdb::Status::Corruption("serialize membership failed");
    }

    s = apply_db_->Put(rocksdb::WriteOptions(), kMemberConfigKey, val);
    if (!s.ok()) {
      return rocksdb::Status::Corruption(
          "record membership in db failed! failed: " + s.ToString());
    }

    for (const auto& member : options_.members) {
      context_->members_.insert(member);
    }
  }

  cmd_service_impl_.reset(new CmdMessageServiceImpl(this));
  rpc_server_.reset(
      new zrpc::RpcServer(loop_.get(), options_.local_ip, options_.local_port));
  rpc_server_->RegisterService(cmd_service_impl_.get());
  rpc_server_->SetThreadNum(kThreadCount);
  rpc_server_->Start();

  int i = 0;
  for (const auto& it : options_.members) {
    std::string ip_port =
        options_.local_ip + ":" + std::to_string(options_.local_port);
    if (ip_port == it) {
      continue;
    }

    AddPeerThread(it);
  }

  {
    std::unique_ptr<zrpc::Thread> primary_thread(new zrpc::Thread());
    zrpc::EventLoop* loop = primary_thread->StartLoop();
    primary_.reset(new Primary(loop, peers_, context_, this, meta_, options_));
    primary_thread_ = std::move(primary_thread);
  }

  {
    std::unique_ptr<zrpc::Thread> apply_thread(new zrpc::Thread());
    zrpc::EventLoop* loop = apply_thread->StartLoop();
    apply_.reset(new Apply(loop, context_, apply_db_, meta_, raft_log_, this));
    apply_thread_ = std::move(apply_thread);
  }

  AddPrimaryTask(kCheckLeader);
  return rocksdb::Status::OK();
}

rocksdb::Status Raft::AddPeer(const std::string& add_server) {
  raft::CmdRequest req;
  BuildAddServerRequest(add_server, &req);
  raft::CmdResponse res;

  rocksdb::Status s = DoCommand(req, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    return rocksdb::Status::OK();
  }

  return rocksdb::Status::Corruption("add_server failed");
}

rocksdb::Status Raft::RemovePeer(const std::string& remove_server) {
  raft::CmdRequest req;
  BuildRemoveServerRequest(remove_server, &req);
  raft::CmdResponse res;

  rocksdb::Status s = DoCommand(req, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    return rocksdb::Status::OK();
  }

  return rocksdb::Status::Corruption("remove server failed");
}

rocksdb::Status Raft::GetAllServers(std::set<std::string>* nodes) {
  raft::CmdRequest req;
  BuildGetAllServersRequest(&req);
  raft::CmdResponse res;
  rocksdb::Status s = DoCommand(req, &res);
  if (!s.ok()) {
    return s;
  }

  if (res.code() == raft::StatusCode::kOk) {
    nodes->clear();
    for (int i = 0; i < res.all_servers().nodes_size(); i++) {
      nodes->insert(res.all_servers().nodes(i));
    }
    return rocksdb::Status::OK();
  }
  return rocksdb::Status::Corruption("getall server failed");
}

rocksdb::Status Raft::CompactRange(rocksdb::CompactRangeOptions options,
                                   const rocksdb::Slice* begin,
                                   const rocksdb::Slice* end) {
  return apply_db_->CompactRange(options, nullptr, nullptr);
}

bool Raft::RemovePeerThread(const std::string& server_addr) {
  LOG_DEBUG << "remove raft peer thread " << options_.local_ip << " "
            << options_.local_port << " " << server_addr;
  if (server_addr ==
      options_.local_ip + ":" + std::to_string(options_.local_port)) {
    return false;
  }

  // std::unique_lock<std::mutex> lk(peer_mutex_);
  for (int i = 0; i < peers_.size(); i++) {
    if (peers_[i]->ServerAddr() == server_addr) {
      peers_.erase(peers_.begin() + i);
      peer_threads_.erase(peer_threads_.begin() + i);
      return true;
    }
  }
  return false;
}

void Raft::AddPeerThread(const std::string& server_addr) {
  if (server_addr ==
      options_.local_ip + ":" + std::to_string(options_.local_port)) {
    return;
  }

  std::string ip;
  int port;
  assert(ParseIpPortString(server_addr, ip, port));
  std::unique_ptr<zrpc::Thread> thread_ptr(new zrpc::Thread());
  zrpc::EventLoop* loop = thread_ptr->StartLoop();
  std::unique_ptr<Peer> peer(new Peer(this, server_addr, ip, port, loop,
                                      context_, meta_, raft_log_, options_));
  peers_.push_back(std::move(peer));
  peer_threads_.push_back(std::move(thread_ptr));

  LOG_DEBUG << "add raft peer thread " << options_.local_ip << " "
            << options_.local_port << " " << server_addr;
}

void Raft::AddPrimaryTask(TaskType type, bool is_delay) {
  primary_->AddTask(type, is_delay);
}

void Raft::Run() { loop_->Run(); }

bool Raft::ParseIpPortString(const std::string& ip_port, std::string& ip,
                             int& port) {
  if (ip_port.empty()) {
    return false;
  }

  size_t pos = ip_port.find(':');
  if (pos == std::string::npos) {
    return false;
  }

  ip = ip_port.substr(0, pos);
  std::string port_str = ip_port.substr(pos + 1);
  int val = 0;
  char* str = port_str.data() + port_str.size();
  val = std::strtoll(port_str.c_str(), &str, 10);
  if ((errno == ERANGE && (val == LLONG_MAX || val == LLONG_MIN)) ||
      (errno != 0 && val == 0)) {
    return false;
  }

  port = val;
  return true;
}

void Raft::UpdatePeerInfo() {
  // std::unique_lock<std::mutex> lk(peer_mutex_);
  for (const auto& it : peers_) {
    // 因为ledaer append rpc后，部分peer更新next_index了（follower宕机)
    // 所以说成为领导节点需要统一更新
    it->SetNextIndex(raft_log_->LastLogIndex() + 1);
    it->SetMatchIndex(0);
  }
}

bool Raft::AdvanceLeaderCommitIndex() {
  uint64_t new_commit_index = QuorumMatchIndex();
  if (context_->commit_index < new_commit_index) {
    context_->commit_index = new_commit_index;
    meta_->SetCommitIndex(context_->commit_index);
    return true;
  }
  return false;
}

void Raft::AdvanceFollowerCommitIndex(uint64_t leader_commit) {
  context_->commit_index = std::min(leader_commit, raft_log_->LastLogIndex());
  meta_->SetCommitIndex(context_->commit_index);
}

bool Raft::IsLeader() {
  if (context_->leader_ip == "" || context_->leader_port == 0) {
    return false;
  }

  if (context_->leader_ip == options_.local_ip &&
      context_->leader_port == options_.local_port) {
    return true;
  }
  return false;
}

bool Raft::GetLeader(std::string* ip, int* port) {
  *ip = context_->leader_ip;
  *port = context_->leader_port;
  return (!ip->empty() && *port != 0);
}

bool Raft::HasLeader() {
  if (context_->role != Role::kLeader || context_->leader_ip == "" ||
      context_->leader_port == 0) {
    return false;
  }
  return true;
}

rocksdb::Status Raft::CreateCheckpoint() {
  return snapshot_->CreateCheckpoint(
      log_db_, path_ + kCheckPointDir + "/" +
                   Env::Default()->TimeToString(Env::Default()->NowMicros()));
}

void Raft::ScheduleApply() { apply_->ScheduleApply(); }

uint64_t Raft::QuorumMatchIndex() {
  std::vector<uint64_t> values;
  // std::unique_lock<std::mutex> lk(peer_mutex_);
  for (const auto& it : peers_) {
    values.push_back(it->MatchIndex());
  }

  std::sort(values.begin(), values.end());
  return values.at(values.size() / 2);
}

bool Raft::DoGetServerStatus(raft::CmdResponse_ServerStatus* res) {
  std::string role_msg;
  switch (context_->role) {
    case Role::kFollower: {
      role_msg = "follower";
      break;
    }

    case Role::kCandidate: {
      role_msg = "candidate";
      break;
    }

    case Role::kLeader: {
      role_msg = "leader";
      break;
    }
  }

  res->set_term(context_->current_term);
  res->set_commit_index(context_->commit_index);
  res->set_role(role_msg);

  std::string ip;
  int port;
  ip = context_->leader_ip;
  port = context_->leader_port;
  if (ip.empty()) {
    res->set_leader_ip("null");
  } else {
    res->set_leader_ip(ip);
  }
  res->set_leader_port(port);

  ip = context_->voted_for_ip;
  port = context_->voted_for_port;
  if (ip.empty()) {
    res->set_voted_for_ip(ip);
  } else {
    res->set_voted_for_ip(ip);
  }
  res->set_voted_for_port(port);

  uint64_t last_log_index;
  uint64_t last_log_term;
  raft_log_->GetLastLogTermAndIndex(&last_log_term, &last_log_index);

  res->set_last_log_term(last_log_term);
  res->set_last_log_index(last_log_index);
  res->set_last_applied(meta_->GetLastApplied());
  return true;
}

bool Raft::GetServerStatus(std::string* msg) {
  raft::CmdResponse_ServerStatus server_status;
  {
    std::unique_lock<std::mutex> lk(context_->global_mutex_);
    DoGetServerStatus(&server_status);
  }

  char str[512];
  snprintf(
      str, sizeof(str),
      "  Node            |    Role    | Term |      Leader      |      "
      "VoteFor      | LastLogTerm | LastLogIdx | CommitIndex | LastApplied |\n"
      "%37s:%-6d%10s%7lu%14s:%-6d%14s:%-6d%10lu%13lu%14lu%13lu     ",
      options_.local_ip.c_str(), options_.local_port,
      server_status.role().c_str(), server_status.term(),
      server_status.leader_ip().c_str(), server_status.leader_port(),
      server_status.voted_for_ip().c_str(), server_status.voted_for_port(),
      server_status.last_log_term(), server_status.last_log_index(),
      server_status.commit_index(), server_status.last_applied());

  msg->clear();
  msg->append(str);
  return true;
}

void Raft::CmdMessageServiceImpl::CmdMessage(
    ::google::protobuf::RpcController* controller,
    const ::raft::CmdRequest* req, ::raft::CmdResponse* res,
    ::google::protobuf::Closure* done) {
  switch (req->type()) {
    case raft::Type::kRequestVote: {
      res->set_type(raft::Type::kRequestVote);
      raft_->ReplyRequestVote(req, res);
      break;
    }

    case raft::Type::kAppendEntries: {
      res->set_type(raft::Type::kAppendEntries);
      raft_->ReplyRequestAppendEntries(req, res);
      break;
    }

    case raft::Type::kRead: {
      res->set_type(raft::Type::kRead);
      res->set_code(raft::StatusCode::kError);
      raft_->DoCommand(*req, res);
      break;
    }

    case raft::Type::kWrite: {
      res->set_type(raft::Type::kWrite);
      res->set_code(raft::StatusCode::kError);
      raft_->DoCommand(*req, res);
      break;
    }

    case raft::Type::kDelete: {
      res->set_type(raft::Type::kDelete);
      res->set_code(raft::StatusCode::kError);
      raft_->DoCommand(*req, res);
      break;
    }

    case raft::Type::kTryLock: {
      res->set_type(raft::Type::kTryLock);
      res->set_code(raft::StatusCode::kError);
      raft_->DoCommand(*req, res);
      break;
    }
    case raft::Type::kUnLock: {
      res->set_type(raft::Type::kUnLock);
      res->set_code(raft::StatusCode::kError);
      raft_->DoCommand(*req, res);
      break;
    }
    case raft::Type::kServerStatus: {
      res->set_type(raft::Type::kServerStatus);
      res->set_code(raft::StatusCode::kError);
      break;
    }
    case raft::Type::kAddServer: {
      res->set_type(raft::Type::kAddServer);
      raft_->DoCommand(*req, res);
      break;
    }
    case raft::Type::kRemoveServer: {
      res->set_type(raft::Type::kRemoveServer);
      raft_->DoCommand(*req, res);
      break;
    }
    case raft::Type::kGetAllServers: {
      res->set_type(raft::Type::kGetAllServers);
      raft_->DoCommand(*req, res);
      break;
    }
    default: {
      res->set_type(raft::Type::kRead);
      LOG_WARN << "unknown cmd type " << req->type();
      break;
    }
  }

  done->Run();
}

static void ReplyResponse(const std::shared_ptr<zrpc::TcpConnection>& conn,
                          const ::google::protobuf::Message& msg) {
  if (conn && conn->Connected()) {
    zrpc::Buffer* buffer = conn->OutputBuffer();
    buffer->AppendInt32(msg.ByteSize());
    buffer->EnsureWritableBytes(msg.ByteSize());

    char* data = buffer->BeginWrite();
    assert(msg.SerializeToArray(data, msg.ByteSize()));

    buffer->HasWritten(msg.ByteSize());
    conn->SendPipe();
  }
}

static void BuildRequestVoteResponse(uint64_t term, bool granted,
                                     raft::CmdResponse* res) {
  res->set_type(raft::Type::kRequestVote);
  raft::CmdResponse_RequestVoteResponse* req_vote_res =
      res->mutable_request_vote_res();
  req_vote_res->set_term(term);
  req_vote_res->set_vote_granted(granted);
}

static void BuildAppendEntriesResponse(bool success, uint64_t term,
                                       uint64_t log_index, uint64_t num_entries,
                                       uint64_t prev_log_index,
                                       uint64_t last_log_term,
                                       raft::CmdResponse* res) {
  res->set_type(raft::Type::kAppendEntries);
  raft::CmdResponse_AppendEntriesResponse* append_entries_res =
      res->mutable_append_entries_res();
  append_entries_res->set_term(term);
  append_entries_res->set_last_log_index(log_index);
  append_entries_res->set_last_log_term(last_log_term);
  append_entries_res->set_prev_log_index(prev_log_index);
  append_entries_res->set_num_entries(num_entries);
  append_entries_res->set_success(success);
}

raft::Entry Raft::BuildEntry(const uint64_t term, const std::string& key,
                             const std::string& value,
                             raft::Entry_OpType type) {
  raft::Entry entry;
  entry.set_key(key);
  entry.set_value(value);
  entry.set_term(term);
  entry.set_optype(type);
  return entry;
}

rocksdb::Status Raft::DeleteDir(const std::string& path) {
  return rocksdb::Env::Default()->DeleteDir(path);
}

void Raft::SetApplyLogCallback(ApplyLogCallback&& callback) {
  apply_->SetApplyLogCallback(std::move(callback));
}

bool Raft::FindPeerSet(const std::string& server_addr) {
  for (const auto& it : peers_) {
    if (server_addr == it->ServerAddr()) {
      return true;
    }
  }
  return false;
}

void Raft::ReplyRequestAppendEntries(const raft::CmdRequest* req,
                                     raft::CmdResponse* res) {
  bool success = false;
  raft::CmdRequest_AppendEntries append_entries = req->append_entries();
  if (append_entries.entries().size() > 0) {
    LOG_DEBUG << "append enrtires : " << append_entries.term() << " "
              << " " << options_.local_ip << " " << options_.local_port;
  } else {
    LOG_DEBUG << "heart beat : " << append_entries.term() << " "
              << options_.local_ip << " " << options_.local_port;
  }

  std::unique_lock<std::mutex> lk(context_->global_mutex_);
  context_->last_op_time = zrpc::TimeStamp::NowMicros();

  if (append_entries.term() < context_->current_term) {
    BuildAppendEntriesResponse(success, context_->current_term,
                               raft_log_->LastLogIndex(),
                               append_entries.entries().size(),
                               append_entries.prev_log_index(), 0, res);
    return;
  } else if ((append_entries.term() > context_->current_term) ||
             (append_entries.term() == context_->current_term &&
              (context_->role == kCandidate ||
               (context_->role == kFollower && context_->leader_ip == "")))) {
    context_->BecomeFollower(append_entries.term(), append_entries.ip(),
                             append_entries.port());
    context_->voted_for_ip = append_entries.ip();
    context_->voted_for_port = append_entries.port();
    meta_->SetCurrentTerm(context_->current_term);
    meta_->SetVotedForIp(context_->voted_for_ip);
    meta_->SetVotedForPort(context_->voted_for_port);
  }

  if (append_entries.prev_log_index() > raft_log_->LastLogIndex()) {
    BuildAppendEntriesResponse(success, context_->current_term,
                               raft_log_->LastLogIndex(),
                               append_entries.entries().size(),
                               append_entries.prev_log_index(), 0, res);
    return;
  }

  if (append_entries.prev_log_index() < raft_log_->LastLogIndex()) {
    LOG_DEBUG << options_.local_ip << ":" << options_.local_port
              << " log_index truncatae suffix prev_log_index "
              << append_entries.prev_log_index() + 1;
    raft_log_->TruncateSuffix(append_entries.prev_log_index() + 1);
  }

  uint64_t my_last_log_term = 0;
  raft::Entry entry;
  if (append_entries.prev_log_index() == 0) {
    my_last_log_term = 0;
  } else if (raft_log_->GetEntry(append_entries.prev_log_index(), &entry)) {
    my_last_log_term = entry.term();
  } else {
    LOG_DEBUG << options_.local_ip << ":" << options_.local_port
              << " can't get entry from raft_log prev_log_index:"
              << append_entries.prev_log_index();
    BuildAppendEntriesResponse(success, context_->current_term,
                               raft_log_->LastLogIndex(),
                               append_entries.entries().size(),
                               append_entries.prev_log_index(), 0, res);
    return;
  }

  if (append_entries.prev_log_term() != my_last_log_term) {
    LOG_DEBUG << options_.local_ip << ":" << options_.local_port
              << " term truncatae suffix prev_log_index "
              << append_entries.prev_log_index();
    raft_log_->TruncateSuffix(append_entries.prev_log_index());
    BuildAppendEntriesResponse(success, context_->current_term,
                               raft_log_->LastLogIndex(),
                               append_entries.entries().size(),
                               append_entries.prev_log_index(), 0, res);
    return;
  }

  std::vector<const raft::Entry*> entries;
  for (int i = 0; i < append_entries.entries().size(); i++) {
    entries.push_back(&append_entries.entries(i));
  }

  uint64_t last_log_term = 0;
  if (append_entries.entries().size() > 0)
    last_log_term =
        append_entries.entries(append_entries.entries().size() - 1).term();
  if (raft_log_->AppendMulti(entries) <= 0) {
    BuildAppendEntriesResponse(
        success, context_->current_term, raft_log_->LastLogIndex(),
        append_entries.entries().size(), append_entries.prev_log_index(),
        last_log_term, res);
    return;
  }

  if (append_entries.leader_commit() != context_->commit_index) {
    LOG_DEBUG << options_.local_ip << ":" << options_.local_port
              << " follower node apply:" << options_.local_ip << " "
              << options_.local_port;
    AdvanceFollowerCommitIndex(append_entries.leader_commit());
    ScheduleApply();
  }

  success = true;
  BuildAppendEntriesResponse(
      success, context_->current_term, raft_log_->LastLogIndex(),
      append_entries.entries().size(), append_entries.prev_log_index(),
      last_log_term, res);
}

void Raft::ReplyRequestVote(const raft::CmdRequest* req,
                            raft::CmdResponse* res) {
  std::unique_lock<std::mutex> lk(context_->global_mutex_);
  bool granted = false;
  raft::CmdRequest_RequestVote req_vote = req->request_vote();
  if (req_vote.term() > context_->current_term) {
    // if (!FindPeerSet(req_vote.ip() + ":" + std::to_string(req_vote.port())))
    // {
    //        BuildRequestVoteResponse(conn, context_->current_term,
    // granted, res); return;
    //}

    context_->BecomeFollower(req_vote.term());
    context_->voted_for_ip.clear();
    context_->voted_for_port = 0;
    meta_->SetCurrentTerm(context_->current_term);
    meta_->SetVotedForIp(context_->voted_for_ip);
    meta_->SetVotedForPort(context_->voted_for_port);
  }

  // 需要注意点, 客户端此时可能已经成为领导节点, 此时回复拒绝，不影响客户端.
  // 成为领导节点 一定是客户端任期>= 当前任期 和 最后日志任期,索引>= 当前最后日志任期,索引
  if (req_vote.term() < context_->current_term) {
    BuildRequestVoteResponse(context_->current_term, granted, res);
    return;
  }

  uint64_t my_last_log_term = 0;
  uint64_t my_last_log_index = 0;
  raft_log_->GetLastLogTermAndIndex(&my_last_log_term, &my_last_log_index);

  if ((req_vote.last_log_term() < my_last_log_term) ||
      ((req_vote.last_log_term() == my_last_log_term) &&
       (req_vote.last_log_index() < my_last_log_index))) {
    BuildRequestVoteResponse(context_->current_term, granted, res);
    return;
  }

  // 已经被某客户端拉取了选票，如果又来了新的客户端，拒绝投票
  if (!context_->voted_for_ip.empty() || context_->voted_for_port != 0) {
    BuildRequestVoteResponse(context_->current_term, granted, res);
    return;
  }

  granted = true;
  context_->voted_for_ip = req_vote.ip();
  context_->voted_for_port = req_vote.port();
  meta_->SetVotedForIp(context_->voted_for_ip);
  meta_->SetVotedForPort(context_->voted_for_port);
  context_->last_op_time = zrpc::TimeStamp::NowMicros();
  BuildRequestVoteResponse(context_->current_term, granted, res);
}

void Raft::ClearLog(uint64_t last_applied_index) {
  std::unique_lock<std::mutex> lk(context_->global_mutex_);
  rocksdb::Status s = raft_log_->TruncatePrefix(last_applied_index - 1);
}

}  // namespace libraft